/**
 * Copyright 2011 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.google.api.adwords.starterkit.Synchronizer;

import com.google.api.adwords.starterkit.db.keys.KeywordKey;
import com.google.api.adwords.starterkit.db.keys.KeywordParamKey;
import com.google.api.adwords.starterkit.db.structures.KeywordParamStructure;
import com.google.api.adwords.starterkit.db.structures.KeywordStructure;
import com.google.api.adwords.starterkit.db.tables.AdGroupTable;
import com.google.api.adwords.starterkit.db.tables.KeywordParamTable;
import com.google.api.adwords.starterkit.db.tables.KeywordTable;
import com.google.api.adwords.starterkit.settings.StarterKitSettings;
import com.google.api.adwords.starterkit.utils.AdWordsReportHandler;
import com.google.api.adwords.starterkit.utils.DumpExceptions;
import com.google.api.adwords.starterkit.utils.RowUtils;
import com.google.api.adwords.starterkit.utils.ServiceUtils;
import com.google.api.adwords.starterkit.utils.Utils;
import com.google.api.adwords.v201101.cm.AdGroupCriterion;
import com.google.api.adwords.v201101.cm.AdGroupCriterionOperation;
import com.google.api.adwords.v201101.cm.AdGroupCriterionReturnValue;
import com.google.api.adwords.v201101.cm.AdGroupCriterionServiceInterface;
import com.google.api.adwords.v201101.cm.AdParam;
import com.google.api.adwords.v201101.cm.AdParamOperation;
import com.google.api.adwords.v201101.cm.AdParamServiceInterface;
import com.google.api.adwords.v201101.cm.ApiError;
import com.google.api.adwords.v201101.cm.ApiException;
import com.google.api.adwords.v201101.cm.Bid;
import com.google.api.adwords.v201101.cm.BiddableAdGroupCriterion;
import com.google.api.adwords.v201101.cm.BudgetOptimizerAdGroupCriterionBids;
import com.google.api.adwords.v201101.cm.ConversionOptimizerAdGroupCriterionBids;
import com.google.api.adwords.v201101.cm.Keyword;
import com.google.api.adwords.v201101.cm.KeywordMatchType;
import com.google.api.adwords.v201101.cm.ManualCPCAdGroupCriterionBids;
import com.google.api.adwords.v201101.cm.ManualCPMAdGroupCriterionBids;
import com.google.api.adwords.v201101.cm.Money;
import com.google.api.adwords.v201101.cm.NegativeAdGroupCriterion;
import com.google.api.adwords.v201101.cm.Operator;
import com.google.api.adwords.v201101.cm.Predicate;
import com.google.api.adwords.v201101.cm.PredicateOperator;
import com.google.api.adwords.v201101.cm.ReportDefinitionReportType;
import com.google.api.adwords.v201101.cm.UserStatus;

import org.apache.log4j.Logger;

import java.rmi.RemoteException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * The Class KeywordSynchronizer is responsible to keep all Keyword 
 * information from AdWords system synchronized with local DB.
 * 
 * @author rodrigo.f.vale@gmail.com (Rodrigo Vale)
 */
public class KeywordSynchronizer {
  private static final Logger log = Logger.getLogger("StarterKit");
  private StarterKitSettings starterKitSetting;
  private ServiceUtils serviceUtils;
  private KeywordParamTable keywordParamTable;
  private AdGroupTable adGroupTable;
  public KeywordTable keywordTable;

  /**
   * Instantiates a new synchronizer.
   * 
   * @param serviceUtils
   *          the service utils to access AdWords system
   * @param starterKitSetting
   *          the starterkit settings
   * @param conn
   *          the connection to local db
   */
  public KeywordSynchronizer(StarterKitSettings starterKitSetting, 
      Connection conn) {
    this.setServiceUtils(new ServiceUtils(starterKitSetting));
    this.starterKitSetting = starterKitSetting;
    
    keywordTable = new KeywordTable(conn);
    keywordParamTable = new KeywordParamTable(conn);
    adGroupTable = new AdGroupTable(conn);
  }

  /**
   * Process updates in keywordParams table.
   */
  protected void putKeywordParams() {
    log.debug("Preparing KwdParams to be sent to AdWords");
    int tries = 0;

    keywordParamTable.updateAdGroupsIds();
    keywordParamTable.updateKeywordsIds();

    while (tries < starterKitSetting.getRetriesOnError()) {
      HashSet<AdParamOperation> operations = new HashSet<AdParamOperation>();
      log.info("Creating/Updating AdParam - Try: " + tries);
      ResultSet rs = null;
      try {
        rs = keywordParamTable.getModified();
        log.debug("Processing Tuples");
        while (rs.next()) {
          KeywordParamStructure kwdParamStruct = keywordParamTable
              .getResultSet(rs);

          // Create ad params.
          if (kwdParamStruct.getParam1().length() > 0) {
            AdParam adParam1 = new AdParam();
            adParam1.setAdGroupId(kwdParamStruct.getAdGroupId());
            adParam1.setCriterionId(kwdParamStruct.getKeywordId());
            adParam1.setInsertionText(kwdParamStruct.getParam1());
            adParam1.setParamIndex(1);

            // Create operations.
            AdParamOperation adParamOperation1 = new AdParamOperation();
            adParamOperation1.setOperand(adParam1);
            adParamOperation1.setOperator(Operator.SET);

            operations.add(adParamOperation1);
          }

          // Create ad params.
          if (kwdParamStruct.getParam2().length() > 0) {
            AdParam adParam2 = new AdParam();
            adParam2.setAdGroupId(kwdParamStruct.getAdGroupId());
            adParam2.setCriterionId(kwdParamStruct.getKeywordId());
            adParam2.setInsertionText(kwdParamStruct.getParam2());
            adParam2.setParamIndex(2);

            // Create operations.
            AdParamOperation adParamOperation2 = new AdParamOperation();
            adParamOperation2.setOperand(adParam2);
            adParamOperation2.setOperator(Operator.SET);

            operations.add(adParamOperation2);
          }
        }
      } catch (SQLException e) {
        log.info("SQL Error processing KeywordParams: " + e.getMessage());
        log.info("StarterKit will try to move on. Some updates may be lost");
      } finally {
        if (rs != null) {
          try {
            rs.close();
          } catch (SQLException e) {
            DumpExceptions.dumpSQLException(e);
          }
        }
      }

      boolean error = processOperationsKeywordParams(operations);
      if (!error) {
        tries = starterKitSetting.getRetriesOnError();
      } else {
        tries++;
        Utils.sleep(starterKitSetting.getSleepOnPolicyError());
      }
    }
    keywordParamTable.resetStatus();
  }

  /**
   * Saves processed updates from KeywordParams to AdWords.
   * 
   * @param operations
   *          The updates to be sent to AdWords
   */
  protected boolean processOperationsKeywordParams(HashSet<AdParamOperation> operations) {
    boolean error = false;
    if (operations.size() > 0) {
      AdParamServiceInterface service = getServiceUtils().getAdParamService();
      List<HashSet<AdParamOperation>> blocks = Utils.split(operations, 1000);

      for (HashSet<AdParamOperation> opBlocks : blocks) {
        AdParamOperation[] op = opBlocks.toArray(new AdParamOperation[opBlocks
            .size()]);
        try {
          service.mutate(op);
          log.debug("Updated " + op.length + " AdParams");
        } catch (ApiException e) {
          DumpExceptions.dumpApiException(e);
          Pattern pattern = Pattern.compile("operations\\[(\\d+)\\].*");
          for (ApiError apiError : e.getErrors()) {
            error = true;
            Matcher matcher = pattern.matcher(apiError.getFieldPath());
            log.debug(apiError.getApiErrorType());
            if (matcher.matches()) {
              int i = Integer.parseInt(matcher.group(1));
              long adGroupId = op[i].getOperand().getAdGroupId();
              long criterionId = op[i].getOperand().getCriterionId();
              int localAdGroupID = adGroupTable.getLocalId(adGroupId);
              int localID = keywordTable.getLocalId(criterionId);
              KeywordParamKey kwdParamKey = new KeywordParamKey(localID,
                  localAdGroupID);
              keywordParamTable.setUpdated(kwdParamKey, false);
              log.debug("Canceling update for AdParam: "
                  + "(localAdGroupID, localID) " + localAdGroupID + ", "
                  + localID);
            }
          }
        } catch (RemoteException e) {
          DumpExceptions.dumpRemoteException(e);
        }
      }
    }
    return error;
  }

  /**
   * Process the updates from keywords table.
   */
  protected void put() {
    log.debug("Preparing Keywords to be sent to AdWords");
    int tries = 0;

    if (starterKitSetting.getKeywordSettings().getAutoPause()) {
      keywordTable.autoPause();
    }
    keywordTable.grantBanned();
    keywordTable.updateAdGroupsIds();

    HashMap<Integer, String> biddingStrategyMap = adGroupTable
        .getCampaignsBiddingStrategy();

    while (tries < starterKitSetting.getRetriesOnError()) {
      HashSet<AdGroupCriterionOperation> operations 
          = new HashSet<AdGroupCriterionOperation>();
      log.info("Creating/Updating Keywords - Try: " + tries);
      ResultSet rs = null;
      try {
        rs = keywordTable.getModified();
        log.debug("Processing Tuples");
        while (rs.next()) {
          KeywordStructure kwStructure = keywordTable.getStructure(rs);
          Keyword keyword = Utils.createKeyword(kwStructure);

          AdGroupCriterionOperation keywordAdGroupCriterionOperation 
              = new AdGroupCriterionOperation();
          if (kwStructure.isNegative()) {
            NegativeAdGroupCriterion negativeAdGroupCriterion 
                = new NegativeAdGroupCriterion(
                kwStructure.getAdGroupId(), null, keyword, null);
            keywordAdGroupCriterionOperation
                .setOperand(negativeAdGroupCriterion);
            if ((kwStructure.getKeywordId() == 0)
                && (starterKitSetting.getKeywordSettings().getAutoCreation())) {
              keywordAdGroupCriterionOperation.setOperator(Operator.ADD);
            } else {
              keyword.setId(kwStructure.getKeywordId());
              // negative keywords can't be paused
              // in that case, "paused" means "to be deleted"
              if (kwStructure.getStatus() == UserStatus.PAUSED) {
                keywordAdGroupCriterionOperation.setOperator(Operator.REMOVE);
              } else {
                keywordAdGroupCriterionOperation.setOperator(Operator.SET);
              }
              keywordAdGroupCriterionOperation.setOperator(Operator.SET);
            }
          } else {
            // Create biddable ad group criterion.
            BiddableAdGroupCriterion keywordBiddableAdGroupCriterion 
                = new BiddableAdGroupCriterion();
            keywordBiddableAdGroupCriterion.setAdGroupId(kwStructure
                .getAdGroupId());
            keywordBiddableAdGroupCriterion.setCriterion(keyword);
            keywordBiddableAdGroupCriterion.setDestinationUrl(kwStructure
                .getDestinationUrl());
            keywordBiddableAdGroupCriterion.setUserStatus(kwStructure
                .getStatus());
            String biddingStrategy = biddingStrategyMap.get(kwStructure
                .getLocalAdGroupId());

            if (biddingStrategy.equals("manualcpc")) {
              ManualCPCAdGroupCriterionBids manualCPCAdGroupCriterionBids 
                  = new ManualCPCAdGroupCriterionBids();
              manualCPCAdGroupCriterionBids.setMaxCpc(new Bid(new Money(null,
                  kwStructure.getMaxCpc())));
              keywordBiddableAdGroupCriterion
                  .setBids(manualCPCAdGroupCriterionBids);
            } else if (biddingStrategy.equals("manualcpm")) {
              ManualCPMAdGroupCriterionBids manualCPMAdGroupCriterionBids 
                  = new ManualCPMAdGroupCriterionBids();
              manualCPMAdGroupCriterionBids.setMaxCpm(new Bid(new Money(null,
                  kwStructure.getMaxCpm())));
              keywordBiddableAdGroupCriterion
                  .setBids(manualCPMAdGroupCriterionBids);
            } else if (biddingStrategy.equals("budgetoptimizer")) {
              BudgetOptimizerAdGroupCriterionBids budgetOptimizer 
                  = new BudgetOptimizerAdGroupCriterionBids();
              keywordBiddableAdGroupCriterion.setBids(budgetOptimizer);
            } else if (biddingStrategy.equals("conversionoptimizer")) {
              ConversionOptimizerAdGroupCriterionBids conversionOptimizer 
                  = new ConversionOptimizerAdGroupCriterionBids();
              keywordBiddableAdGroupCriterion.setBids(conversionOptimizer);
            }

            // Create operations.
            keywordAdGroupCriterionOperation
                .setOperand(keywordBiddableAdGroupCriterion);
            if (kwStructure.getKeywordId() == 0) {
              keywordAdGroupCriterionOperation.setOperator(Operator.ADD);
            } else {
              keyword.setId(kwStructure.getKeywordId());
              keywordAdGroupCriterionOperation.setOperator(Operator.SET);
            }
          }
          operations.add(keywordAdGroupCriterionOperation);
        }
        rs.close();
      } catch (SQLException e3) {
        log.debug("SQL Error on saveCampaigns. Error: " + e3.getMessage());
        log.debug("StarterKit will try to move on. Some updates may be lost.");
      } finally {
        if (rs != null) {
          try {
            rs.close();
          } catch (SQLException e) {
            DumpExceptions.dumpSQLException(e);
          }
        }
      }

      boolean error = processOperations(operations);
      // If no policy error we don't need to run the process again
      if (!error) {
        tries = starterKitSetting.getRetriesOnError();
      } else {
        Utils.sleep(starterKitSetting.getSleepOnPolicyError());
        tries++;
      }
    }
    keywordTable.resetIsBanned(true);
  }

  /**
   * Saves the processed updates from keywords table to AdWords.
   * 
   * @param operations
   *          The updates to be sent to AdWords
   */
  protected boolean processOperations(
      HashSet<AdGroupCriterionOperation> operations) {
    boolean error = false;
    if (operations.size() > 0) {
      log.info("Processing: " + operations.size());
      AdGroupCriterionServiceInterface adGroupCriterionService 
          = getServiceUtils().getAdGroupCriterionService();
      List<HashSet<AdGroupCriterionOperation>> blocks 
          = Utils.split(operations, 1000);
      for (HashSet<AdGroupCriterionOperation> opBlocks : blocks) {
        log.info("Processing Block: " + opBlocks.size());
        AdGroupCriterionOperation[] op = opBlocks
            .toArray(new AdGroupCriterionOperation[opBlocks.size()]);

        // Add ad group criteria.
        AdGroupCriterionReturnValue result = null;
        try {
          result = adGroupCriterionService.mutate(op);
          // Display ad group criteria.
          if (result != null && result.getValue() != null) {
            for (AdGroupCriterion adGroupCriterion : result.getValue()) {
              Keyword keyword = (Keyword) adGroupCriterion.getCriterion();
              long adGroupId = adGroupCriterion.getAdGroupId();
              int localAdGroupId = adGroupTable.getLocalId(adGroupId);
              String text = keyword.getText();
              KeywordMatchType matchType = keyword.getMatchType();
              KeywordKey keywordKey = new KeywordKey(localAdGroupId, text,
                  matchType);
              keywordTable.addBatchKeywordId(keywordKey, keyword.getId(),
                  false, false);
            }
            keywordTable.executeBatch();
          }
          log.debug("Updated " + op.length + " Keywords");
        } catch (ApiException e1) {
          error = true;
          Pattern pattern = Pattern.compile("operations\\[(\\d+)\\].*");
          log.debug(e1.getErrors().length + " ApiExceptions. Removing it.");
          int count = 0;
          HashMap<String, Integer> errorsMsg = new HashMap<String, Integer>();
          HashMap<String, String> errorsExample = new HashMap<String, String>();
          for (ApiError apiError : e1.getErrors()) {
            String errorKey = apiError.getErrorString();
            int value = 0;
            if (errorsMsg.containsKey(errorKey)) {
              value = errorsMsg.get(errorKey);
            }
            value++;
            errorsMsg.put(errorKey, value);
            errorsExample.put(errorKey, apiError.getTrigger());
            Matcher matcher = pattern.matcher(apiError.getFieldPath());
            if (matcher.matches()) {
              count++;
              int i = Integer.parseInt(matcher.group(1));
              AdGroupCriterion adGroupCriterion = op[i].getOperand();
              Keyword keyword = (Keyword) adGroupCriterion.getCriterion();
              long adGroupId = adGroupCriterion.getAdGroupId();
              int localAdGroupId = adGroupTable.getLocalId(adGroupId);
              String text = keyword.getText();
              KeywordMatchType matchType = keyword.getMatchType();
              KeywordKey keywordKey = new KeywordKey(localAdGroupId, text,
                  matchType);
              keywordTable.addBatchUpdated(keywordKey, false, false, apiError
                  .getErrorString(), apiError.getTrigger());
            }
          }
          Iterator<String> i = errorsMsg.keySet().iterator();
          while (i.hasNext()) {
            String key = (String) i.next();
            String trigger = errorsExample.get(key);
            int value = errorsMsg.get(key);
            log.debug("Error: " + key + " [" + value + "] - Trigger Example: "
                + trigger);
            if (key.equals("InternalApiError."
                + "UNEXPECTED_INTERNAL_API_ERROR")) {
              Utils.sleep(starterKitSetting.getSleepOnInternalError());
            }
          }
          log.debug("Removed " + count + " ApiExceptions");
          keywordTable.executeBatch();
        } catch (RemoteException e) {
          error = true;
          DumpExceptions.dumpRemoteException(e);
          Utils.sleep(starterKitSetting.getSleepOnInternalError());
        }
      }
    }
    return error;
  }

  /**
   * Load keywords performance from adwords.
   * 
   */
  public void getStats() {
    log.debug("Processing AdKeywords Performance JSON Objects");

    keywordTable.resetKeywordsStats();

    AdWordsReportHandler adWordsReportHandler = getReportHandlerPerformance();
    while (adWordsReportHandler.hasNext()) {
      HashMap<String, String> row = adWordsReportHandler.next();
      if (row.containsKey("ctr")) {
        row.put("ctr", row.get("ctr").replace("%", ""));
      }
      float ctr = (float) RowUtils.getDouble(row, "ctr", 0);
      long clicks = RowUtils.getLong(row, "clicks", 0);
      float cost = (float) RowUtils.getDouble(row, "cost", 0);
      long imp = RowUtils.getLong(row, "impressions", 0);
      long conv1PerClick = RowUtils.getLong(row, "conv1PerClick", 0);
      long convManyPerClick = RowUtils.getLong(row, "convManyPerClick", 0);
      float avgPosition = (float) RowUtils.getDouble(row, "avgPosition", 0);
      float adGroupMaxCpa = 0;
      float costConv1PerClick = 
          (float) RowUtils.getDouble(row, "costConv1PerClick", 0);
      float costConvManyPerClick = 
          (float) RowUtils.getDouble(row, "costConvManyPerClick", 0);
      float firstPageCPC = 
          (float) RowUtils.getDouble(row, "firstPageCPC", 0);
      long keywordId = RowUtils.getLong(row, "keywordID", 0);
      long adGroupId = RowUtils.getLong(row, "adGroupID", 0);

      keywordTable.addStatsToBatch(ctr, clicks, cost, imp,
          conv1PerClick, convManyPerClick, avgPosition, adGroupMaxCpa,
          costConv1PerClick, costConvManyPerClick, firstPageCPC,
          keywordId, adGroupId);
    }
    keywordTable.executeStatsBatch();
  }

  /**
   * Load keyword param from adwords.
   */
  private void getKeywordParam() {
    // there is no need for this method. Params can only be modified by
    // the API. So the DB will be always sync.
  
    keywordParamTable.resetStatus();
  }

  /**
   * Process a specific keyword from adwords report and save it into DB.
   * 
   * @param row
   *          the AdWords report row
   * @param adGroupIdsMap
   *          the adgroup ids map
   * @throws SQLException
   *           the sQL exception
   */
  private void processReportData(HashMap<String, String> row,
      HashMap<Long, Integer> adGroupIdsMap) throws SQLException {

    long keywordId = RowUtils.getLong(row, "keywordID", -1);

    KeywordStructure kwdStructure = new KeywordStructure();
    kwdStructure.setAdGroupId(RowUtils.getLong(row, "adGroupID", 0));
    String match = RowUtils.getString(row, "matchType", "").toUpperCase();
    kwdStructure.setMatchType(KeywordMatchType.fromString(match));
    kwdStructure.setText(RowUtils.getString(row, "keyword", ""));
    String status = RowUtils.getString(row, "keywordState", "").toUpperCase();

    kwdStructure.setStatus(KeywordStructure.normalizeStatus(status));
    
    double dMaxCPC = RowUtils.getDouble(row, "maxCPC", 0);
    long maxCPC = (long) (dMaxCPC * 1000000);
    kwdStructure.setMaxCpc(maxCPC);
    
    double dMaxCpm = RowUtils.getLong(row, "maxCPM", 0);
    long maxCpm = (long) (dMaxCpm * 1000000);
    kwdStructure.setMaxCpm(maxCpm);
    
    kwdStructure.setDestinationUrl(RowUtils.getString(row, "destinationURL",
        ""));
    if (kwdStructure.getDestinationUrl().equals("defaultURL")) {
      kwdStructure.setDestinationUrl("");
    }

    kwdStructure.setNegative(RowUtils.getBoolean(row, "isNegative", false));
    kwdStructure.setKeywordId(keywordId);

    String agStatus = RowUtils.getString(row, "adGroupState", "");
    if (!agStatus.equals("deleted")
        && (!kwdStructure.getStatus().equals("deleted"))) {
      if (adGroupIdsMap.containsKey(kwdStructure.getAdGroupId())) {
        kwdStructure.setLocalAdGroupId(adGroupIdsMap.get(kwdStructure
            .getAdGroupId()));
        KeywordKey keywordKey = new KeywordKey(
            kwdStructure.getLocalAdGroupId(), kwdStructure.getText(),
            kwdStructure.getMatchType());

        KeywordStructure kwdStructDB = keywordTable.get(keywordKey);

        if (kwdStructDB == null) {
          // keywords created outside of StarterKit can't be modified
          // setting canModify flags to false
          kwdStructure.setCpcModifiable(false);
          kwdStructure.setMatchModifiable(false);
          kwdStructure.setStatusModifiable(false);
          kwdStructure.setUrlModifiable(false);

          // also, it can't be banned because it exists
          kwdStructure.setBanned(false);
          keywordTable.addToBatch(kwdStructure, false, false);
        } else {
          kwdStructure.setLocalId(kwdStructDB.getLocalId());
          kwdStructure.setCpcModifiable(kwdStructDB.isCpcModifiable());
          kwdStructure.setMatchModifiable(kwdStructDB.isMatchModifiable());
          kwdStructure.setStatusModifiable(kwdStructDB.isStatusModifiable());
          kwdStructure.setUrlModifiable(kwdStructDB.isUrlModifiable());

          if (!kwdStructure.getStatus().equals(kwdStructDB.getStatus())) {
            kwdStructure.setStatusModifiable(false);
          }

          if (kwdStructure.getMaxCpc() != kwdStructDB.getMaxCpc()) {
            kwdStructure.setCpcModifiable(false);
          }
          if (kwdStructure.getMaxCpm() != kwdStructDB.getMaxCpm()) {
            kwdStructure.setCpcModifiable(false);
          }

          if (!kwdStructure.getMatchType().equals(kwdStructDB.getMatchType())) {
            kwdStructure.setMatchModifiable(false);
          }

          // ok keywords exists at DB and at AdWords, so it won't be
          // banned
          kwdStructure.setBanned(false);
          keywordTable.addToBatch(kwdStructure, false, false);
        }
      }
    }
  }

  /**
   * Load keywords from AdWords.
   * 
   */
  public void get() {
    log.debug("Processing Keywords Report");
    AdWordsReportHandler adWordsReportHandler = getReportHandler();

    HashMap<Long, Integer> adGroupIdsMap = adGroupTable.getAdGroupIdsMap();

    // all old keywords will be banned by default. Keywords that exists
    // at rows will be set to unbanned during nexts steps
    keywordTable.resetIsBanned(true);
    
    try {
      int count = 0;
      while (adWordsReportHandler.hasNext()) {
        HashMap<String, String> row = adWordsReportHandler.next();
        processReportData(row, adGroupIdsMap);
        count++;
        if (count == 100000) {
          count = 0;
          keywordTable.executeBatch();
        }
      }
      keywordTable.executeBatch();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }
    keywordTable.resetIsUpdated(false);
    keywordTable.resetIsTouched(false);

    keywordTable.resetStatus();
    getKeywordParam();
  }

  /**
   * Gets the keyword structure report.
   * 
   * @return the keyword structure report
   */
  public AdWordsReportHandler getReportHandler() {
    log.debug("Defining Keyword Structure Report");

    String fields[] = new String[] { "CampaignId", "CampaignName", "AdGroupId",
        "AdGroupName", "KeywordMatchType", "Id", "KeywordText", "Status",
        "MaxCpm", "MaxCpc", "DestinationUrl", "IsNegative", "AdGroupStatus" };

    String fileName = starterKitSetting.getCacheDir().concat(
        "KeywordStrututureReport.xml");

    String campaignStatus[] = null;
    if (starterKitSetting.getDownloadDeletedCampaigns()) {
      campaignStatus = new String[] { "ACTIVE", "PAUSED", "DELETED" };
    } else {
      campaignStatus = new String[] { "ACTIVE", "PAUSED" };
    }

    // Create campaing predicate.
    Predicate campaingPredicate = new Predicate();
    campaingPredicate.setField("CampaignStatus");
    campaingPredicate.setOperator(PredicateOperator.IN);
    campaingPredicate.setValues(campaignStatus);

    Predicate[] predicates = new Predicate[] { campaingPredicate };

    AdWordsReportHandler adWordsReportHandler = new AdWordsReportHandler(
        getServiceUtils(), starterKitSetting);
    adWordsReportHandler.requestAdWordsReport(
        ReportDefinitionReportType.KEYWORDS_PERFORMANCE_REPORT, fields,
        fileName, predicates);

    return adWordsReportHandler;
  }

  /**
   * Gets the keyword performance report.
   * 
   * @return the keyword performance report
   */
  public AdWordsReportHandler getReportHandlerPerformance() {
    log.debug("Defining Keyword Performance Report");

    String fields[] = new String[] { "AdGroupId", "Id", "AverageCpc", "Ctr",
        "Clicks", "Cost", "Impressions", "Conversions",
        "ConversionsManyPerClick", "AveragePosition", "CostPerConversion",
        "CostPerConversionManyPerClick", "FirstPageCpc" };

    String fileName = starterKitSetting.getCacheDir().concat(
        "KeywordPerformanceReport.xml");

    String campaignStatus[] = null;
    if (starterKitSetting.getDownloadDeletedCampaigns()) {
      campaignStatus = new String[] { "ACTIVE", "PAUSED", "DELETED" };
    } else {
      campaignStatus = new String[] { "ACTIVE", "PAUSED" };
    }
    // Create campaing predicate.
    Predicate campaingPredicate = new Predicate();
    campaingPredicate.setField("CampaignStatus");
    campaingPredicate.setOperator(PredicateOperator.IN);
    campaingPredicate.setValues(campaignStatus);

    Predicate[] predicates = new Predicate[] { campaingPredicate };

    AdWordsReportHandler adWordsReportHandler = new AdWordsReportHandler(
        getServiceUtils(), starterKitSetting);
    adWordsReportHandler.requestAdWordsReport(
        ReportDefinitionReportType.KEYWORDS_PERFORMANCE_REPORT, fields,
        fileName, predicates);

    return adWordsReportHandler;
  }

  public void setServiceUtils(ServiceUtils serviceUtils) {
    this.serviceUtils = serviceUtils;
  }

  public ServiceUtils getServiceUtils() {
    return serviceUtils;
  }
}